package io.renren.kafka.consume;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import io.renren.htby.utils.KafkaConsumeUtils;

/**
 * 心跳监控
 * 
 * @author lfy.xys
 * @date 2018年5月31日
 *
 */
@Component
public class HeartbeatConsumer {

	private KafkaConsumer<String, String> kafkaConsumer = null;
	
	private static Logger logger = LoggerFactory.getLogger(HeartbeatConsumer.class);
	
	public void setConsumer() {
		Properties props = new Properties();
		// bootstrap.servers 和Producer一样，是指向Kafka集群的IP地址，以逗号分隔。
		props.put("bootstrap.servers", KafkaConsumeUtils.bootstrapServers);
		props.put("group.id", KafkaConsumeUtils.hearbeat_group_id);//使用 监控的组
		props.put("enable.auto.commit", KafkaConsumeUtils.enable_auto_commit);
		props.put("auto.commit.interval.ms", KafkaConsumeUtils.auto_commit_interval_ms);
		props.put("auto.offset.reset", KafkaConsumeUtils.auto_offset_reset);
		props.put("session.timeout.ms", KafkaConsumeUtils.session_timeout_ms);
		// 发序列化。Consumer把来自Kafka集群的二进制消息反序列化为指定的类型。因本例中的Producer使用的是String类型，所以调用StringDeserializer来反序列化
		props.put("key.deserializer", KafkaConsumeUtils.key_deserializer);
		props.put("value.deserializer", KafkaConsumeUtils.value_deserializer);
		kafkaConsumer = new KafkaConsumer<String, String>(props);
	}

	public void consume(String topic) {
		if (kafkaConsumer == null) {
			this.setConsumer();
		}
		// Consumer订阅了Topic的消息，
		kafkaConsumer.subscribe(Arrays.asList(topic));
		logger.info("订阅了topic:{}",topic);
		while (true) {
			// Consumer调用poll方法来轮循Kafka集群的消息，其中的参数100是超时时间（Consumer等待直到Kafka集群中没有消息为止）：
			ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
			for (ConsumerRecord<String, String> record : records) {
				String msg = record.value();
//				logger.info("--------------- offset ={}", record.offset());
				logger.info("--------------------得到心跳信息 ：{}", msg);
				//TODO 消费的数据 业务未知
			}
		}
	}
}
